/*-------------------------------------------------------------------------
 *
 * pg_publication.c
 *		出版物C API操作
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *		src/backend/catalog/pg_publication.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "access/genam.h"
#include "access/heapam.h"
#include "access/htup_details.h"
#include "access/tableam.h"
#include "access/xact.h"
#include "catalog/catalog.h"
#include "catalog/dependency.h"
#include "catalog/index.h"
#include "catalog/indexing.h"
#include "catalog/namespace.h"
#include "catalog/partition.h"
#include "catalog/objectaccess.h"
#include "catalog/objectaddress.h"
#include "catalog/pg_inherits.h"
#include "catalog/pg_namespace.h"
#include "catalog/pg_publication.h"
#include "catalog/pg_publication_namespace.h"
#include "catalog/pg_publication_rel.h"
#include "catalog/pg_type.h"
#include "commands/publicationcmds.h"
#include "funcapi.h"
#include "miscadmin.h"
#include "utils/array.h"
#include "utils/builtins.h"
#include "utils/catcache.h"
#include "utils/fmgroids.h"
#include "utils/inval.h"
#include "utils/lsyscache.h"
#include "utils/rel.h"
#include "utils/syscache.h"

#ifdef FDDM
#include "utils/varlena.h"

#define FDD_EXCLUDE_SCHEMA_CONF "fdd.exclude_schema"

static List* fdd_get_exclude_schema(char *fc_fmmrExcludeSchemaStr);
#endif

static void fc_publication_translate_columns(Relation fc_targetrel, List *fc_columns,
										  int *fc_natts, AttrNumber **fc_attrs);

/* 
 * 检查关系是否可以在给定发布中，并在不满足时抛出适当错误。
 */
static void fc_check_publication_add_relation(Relation fc_targetrel)
{
	/* 必须是常规或分区表 */
	if (RelationGetForm(fc_targetrel)->relkind != RELKIND_RELATION &&
		RelationGetForm(fc_targetrel)->relkind != RELKIND_PARTITIONED_TABLE)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("cannot add relation \"%s\" to publication",
						RelationGetRelationName(fc_targetrel)),
				 errdetail_relkind_not_supported(RelationGetForm(fc_targetrel)->relkind)));

	/* 不能是系统表 */
	if (IsCatalogRelation(fc_targetrel))
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("cannot add relation \"%s\" to publication",
						RelationGetRelationName(fc_targetrel)),
				 errdetail("This operation is not supported for system tables.")));

	/* UNLOGGED和TEMP关系不能作为发布的一部分。 */
	if (fc_targetrel->rd_rel->relpersistence == RELPERSISTENCE_TEMP)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("cannot add relation \"%s\" to publication",
						RelationGetRelationName(fc_targetrel)),
				 errdetail("This operation is not supported for temporary tables.")));
	else if (fc_targetrel->rd_rel->relpersistence == RELPERSISTENCE_UNLOGGED)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("cannot add relation \"%s\" to publication",
						RelationGetRelationName(fc_targetrel)),
				 errdetail("This operation is not supported for unlogged tables.")));
}

/* 
 * 检查方案是否可以在给定发布中，并在不满足时抛出适当错误。
 */
static void fc_check_publication_add_schema(Oid fc_schemaid)
{
	/* 不能是系统命名空间 */
	if (IsCatalogNamespace(fc_schemaid) || IsToastNamespace(fc_schemaid))
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("cannot add schema \"%s\" to publication",
						get_namespace_name(fc_schemaid)),
				 errdetail("This operation is not supported for system schemas.")));

	/* 不能是临时命名空间 */
	if (isAnyTempNamespace(fc_schemaid))
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("cannot add schema \"%s\" to publication",
						get_namespace_name(fc_schemaid)),
				 errdetail("Temporary schemas cannot be replicated.")));
}

/*
 * 返回由 oid 和 Form_pg_class 条目表示的关系是否可发布。
 *
 * 与上述检查相同，但不需要打开关系，并且也不会抛出错误。
 *
 * XXX 这还排除了所有 relid < FirstNormalObjectId 的表，
 * 也就是说，所有在 initdb 期间创建的表。这主要影响预安装的
 * information_schema。IsCatalogRelationOid() 仅排除了 relid <
 * FirstUnpinnedObjectId 的表，使得该测试相当冗余，
 * 但我们确实应该去掉 FirstNormalObjectId 测试而不是
 * IsCatalogRelationOid。我们今天无法做到这一点，因为我们不想
 * 让 information_schema 表被视为可发布；但这个测试
 * 实际上对于这个目的不够充分，因为 information_schema 可能被
 * 删除并重新加载，然后它将被视为可发布。最佳的长期解决方案
 * 可能是向 pg_class 添加一个 "relispublishable" 布尔值，
 * 并依赖于它而不是 OID 检查。
 */
static bool fc_is_publishable_class(Oid fc_relid, Form_pg_class fc_reltuple)
{
	return (fc_reltuple->relkind == RELKIND_RELATION ||
			fc_reltuple->relkind == RELKIND_PARTITIONED_TABLE) &&
		!IsCatalogRelationOid(fc_relid) &&
		fc_reltuple->relpersistence == RELPERSISTENCE_PERMANENT &&
		fc_relid >= FirstNormalObjectId;
}

/*
 * 过滤掉其父表也在发布中指定的分区。
 */
static List * fc_filter_partitions(List *fc_relids)
{
	List	   *fc_result = NIL;
	ListCell   *fc_lc;
	ListCell   *fc_lc2;

	foreach(fc_lc, fc_relids)
	{
		bool		fc_skip = false;
		List	   *fc_ancestors = NIL;
		Oid			fc_relid = lfirst_oid(fc_lc);

		if (get_rel_relispartition(fc_relid))
			fc_ancestors = get_partition_ancestors(fc_relid);

		foreach(fc_lc2, fc_ancestors)
		{
			Oid			fc_ancestor = lfirst_oid(fc_lc2);

			/* 检查父表是否存在于发布的表列表中。 */
			if (list_member_oid(fc_relids, fc_ancestor))
			{
				fc_skip = true;
				break;
			}
		}

		if (!fc_skip)
			fc_result = lappend_oid(fc_result, fc_relid);
	}

	return fc_result;
}

/*
 * 另一种变体，使用 Relation。
 */
bool is_publishable_relation(Relation fc_rel)
{
	return fc_is_publishable_class(RelationGetRelid(fc_rel), fc_rel->rd_rel);
}

/*
 * 如果与发布关联的架构存在，则返回 true；如果没有与发布关联的架构，则返回 false。
 */
bool is_schema_publication(Oid fc_pubid)
{
	Relation	fc_pubschsrel;
	ScanKeyData fc_scankey;
	SysScanDesc fc_scan;
	HeapTuple	fc_tup;
	bool		fc_result = false;

	fc_pubschsrel = table_open(PublicationNamespaceRelationId, AccessShareLock);
	ScanKeyInit(&fc_scankey,
				Anum_pg_publication_namespace_pnpubid,
				BTEqualStrategyNumber, F_OIDEQ,
				ObjectIdGetDatum(fc_pubid));

	fc_scan = systable_beginscan(fc_pubschsrel,
							  PublicationNamespacePnnspidPnpubidIndexId,
							  true, NULL, 1, &fc_scankey);
	fc_tup = systable_getnext(fc_scan);
	fc_result = HeapTupleIsValid(fc_tup);

	systable_endscan(fc_scan);
	table_close(fc_pubschsrel, AccessShareLock);

	return fc_result;
}

/*
 * 以上内容的 SQL 可调用变体
 *
 * 当关系不存在时返回 null。这是为了在 psql 中避免在并发目录更改时产生不必要的错误而设计的。
 */
Datum pg_relation_is_publishable(PG_FUNCTION_ARGS)
{
	Oid			fc_relid = PG_GETARG_OID(0);
	HeapTuple	fc_tuple;
	bool		fc_result;

	fc_tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(fc_relid));
	if (!HeapTupleIsValid(fc_tuple))
		PG_RETURN_NULL();
	fc_result = fc_is_publishable_class(fc_relid, (Form_pg_class) GETSTRUCT(fc_tuple));
	ReleaseSysCache(fc_tuple);
	PG_RETURN_BOOL(fc_result);
}

/*
 * 根据指定关系的发布分区选项获取关系。
 */
List * GetPubPartitionOptionRelations(List *fc_result, PublicationPartOpt fc_pub_partopt,
							   Oid fc_relid)
{
	if (get_rel_relkind(fc_relid) == RELKIND_PARTITIONED_TABLE &&
		fc_pub_partopt != PUBLICATION_PART_ROOT)
	{
		List	   *fc_all_parts = find_all_inheritors(fc_relid, NoLock,
													NULL);

		if (fc_pub_partopt == PUBLICATION_PART_ALL)
			fc_result = list_concat(fc_result, fc_all_parts);
		else if (fc_pub_partopt == PUBLICATION_PART_LEAF)
		{
			ListCell   *fc_lc;

			foreach(fc_lc, fc_all_parts)
			{
				Oid			fc_partOid = lfirst_oid(fc_lc);

				if (get_rel_relkind(fc_partOid) != RELKIND_PARTITIONED_TABLE)
					fc_result = lappend_oid(fc_result, fc_partOid);
			}
		}
		else
			Assert(false);
	}
	else
		fc_result = lappend_oid(fc_result, fc_relid);

	return fc_result;
}

/*
 * 如果有的话，返回通过此发布发布的最顶层祖先的 relid，并将其祖先级别设置为 ancestor_level，
 * 否则返回 InvalidOid。
 *
 * ancestor_level 值允许我们比较多个发布的结果，并决定哪个值更高。
 *
 * 请注意，祖先列表应该按顺序排列，使得最顶层祖先位于列表的末尾。
 */
Oid GetTopMostAncestorInPublication(Oid fc_puboid, List *fc_ancestors, int *fc_ancestor_level)
{
	ListCell   *fc_lc;
	Oid			fc_topmost_relid = InvalidOid;
	int			fc_level = 0;

	/*
	 * 找到在此发布中的“最顶层”祖先。
	 */
	foreach(fc_lc, fc_ancestors)
	{
		Oid			fc_ancestor = lfirst_oid(fc_lc);
		List	   *fc_apubids = GetRelationPublications(fc_ancestor);
		List	   *fc_aschemaPubids = NIL;

		fc_level++;

		if (list_member_oid(fc_apubids, fc_puboid))
		{
			fc_topmost_relid = fc_ancestor;

			if (fc_ancestor_level)
				*fc_ancestor_level = fc_level;
		}
		else
		{
			fc_aschemaPubids = GetSchemaPublications(get_rel_namespace(fc_ancestor));
			if (list_member_oid(fc_aschemaPubids, fc_puboid))
			{
				fc_topmost_relid = fc_ancestor;

				if (fc_ancestor_level)
					*fc_ancestor_level = fc_level;
			}
		}

		list_free(fc_apubids);
		list_free(fc_aschemaPubids);
	}

	return fc_topmost_relid;
}

/*
 * 插入新的发布/关系映射。
 */
ObjectAddress publication_add_relation(Oid fc_pubid, PublicationRelInfo *fc_pri,
						 bool fc_if_not_exists)
{
	Relation	fc_rel;
	HeapTuple	fc_tup;
	Datum		fc_values[Natts_pg_publication_rel];
	bool		fc_nulls[Natts_pg_publication_rel];
	Relation	fc_targetrel = fc_pri->relation;
	Oid			fc_relid = RelationGetRelid(fc_targetrel);
	Oid			fc_pubreloid;
	Publication *fc_pub = GetPublication(fc_pubid);
	AttrNumber *fc_attarray = NULL;
	int			fc_natts = 0;
	ObjectAddress fc_myself,
				fc_referenced;
	List	   *fc_relids = NIL;

	fc_rel = table_open(PublicationRelRelationId, RowExclusiveLock);

	/*
	 * 检查重复项。请注意，这实际上并不能真正防止重复项，它仅用于在常见情况下提供更好的错误信息。
	 * 真实的保护在于目录上的唯一键。
	 */
	if (SearchSysCacheExists2(PUBLICATIONRELMAP, ObjectIdGetDatum(fc_relid),
							  ObjectIdGetDatum(fc_pubid)))
	{
		table_close(fc_rel, RowExclusiveLock);

		if (fc_if_not_exists)
			return InvalidObjectAddress;

		ereport(ERROR,
				(errcode(ERRCODE_DUPLICATE_OBJECT),
				 errmsg("relation \"%s\" is already member of publication \"%s\"",
						RelationGetRelationName(fc_targetrel), fc_pub->name)));
	}

	fc_check_publication_add_relation(fc_targetrel);

	/*
	 * 将列名转换为 attnums，并确保列列表只包含允许的元素（没有系统或生成的列等）。
	 * 同时构建一个 attnums 数组，用于存储在目录中。
	 */
	fc_publication_translate_columns(fc_pri->relation, fc_pri->columns,
								  &fc_natts, &fc_attarray);

	/* 形成一个元组。 */
	memset(fc_values, 0, sizeof(fc_values));
	memset(fc_nulls, false, sizeof(fc_nulls));

	fc_pubreloid = GetNewOidWithIndex(fc_rel, PublicationRelObjectIndexId,
								   Anum_pg_publication_rel_oid);
	fc_values[Anum_pg_publication_rel_oid - 1] = ObjectIdGetDatum(fc_pubreloid);
	fc_values[Anum_pg_publication_rel_prpubid - 1] =
		ObjectIdGetDatum(fc_pubid);
	fc_values[Anum_pg_publication_rel_prrelid - 1] =
		ObjectIdGetDatum(fc_relid);

	/* 如果可用，添加资格 */
	if (fc_pri->whereClause != NULL)
		fc_values[Anum_pg_publication_rel_prqual - 1] = CStringGetTextDatum(nodeToString(fc_pri->whereClause));
	else
		fc_nulls[Anum_pg_publication_rel_prqual - 1] = true;

	/* 如果可用，添加列列表 */
	if (fc_pri->columns)
		fc_values[Anum_pg_publication_rel_prattrs - 1] = PointerGetDatum(buildint2vector(fc_attarray, fc_natts));
	else
		fc_nulls[Anum_pg_publication_rel_prattrs - 1] = true;

	fc_tup = heap_form_tuple(RelationGetDescr(fc_rel), fc_values, fc_nulls);

	/* 将元组插入目录。 */
	CatalogTupleInsert(fc_rel, fc_tup);
	heap_freetuple(fc_tup);

	/* 根据需要注册依赖关系 */
	ObjectAddressSet(fc_myself, PublicationRelRelationId, fc_pubreloid);

	/* 添加对发布的依赖 */
	ObjectAddressSet(fc_referenced, PublicationRelationId, fc_pubid);
	recordDependencyOn(&fc_myself, &fc_referenced, DEPENDENCY_AUTO);

	/* 添加对关系的依赖 */
	ObjectAddressSet(fc_referenced, RelationRelationId, fc_relid);
	recordDependencyOn(&fc_myself, &fc_referenced, DEPENDENCY_AUTO);

	/* 添加对资格中提到的对象的依赖 */
	if (fc_pri->whereClause)
		recordDependencyOnSingleRelExpr(&fc_myself, fc_pri->whereClause, fc_relid,
										DEPENDENCY_NORMAL, DEPENDENCY_NORMAL,
										false);

	/* 如果列有列出，添加对列的依赖 */
	for (int fc_i = 0; fc_i < fc_natts; fc_i++)
	{
		ObjectAddressSubSet(fc_referenced, RelationRelationId, fc_relid, fc_attarray[fc_i]);
		recordDependencyOn(&fc_myself, &fc_referenced, DEPENDENCY_NORMAL);
	}

	/* 关闭表。 */
	table_close(fc_rel, RowExclusiveLock);

	/*
	 * 使relcache失效，以便重建发布信息。
	 *
	 * 对于分区表，我们必须使所包含的所有分区失效
	 * 在各自的分区层次结构中，而不仅仅是显式
	 * 在发布中提到的那个。这是必需的，因为当发布父表时，
	 * 我们隐式地发布子表。
	 */
	fc_relids = GetPubPartitionOptionRelations(fc_relids, PUBLICATION_PART_ALL,
											fc_relid);

	InvalidatePublicationRels(fc_relids);

	return fc_myself;
}

/* attnums 的 qsort 比较器 */
static int fc_compare_int16(const void *fc_a, const void *fc_b)
{
	int			fc_av = *(const int16 *) fc_a;
	int			fc_bv = *(const int16 *) fc_b;

	/* 如果 int 的宽度大于 int16，这个不会溢出 */
	return (fc_av - fc_bv);
}

/*
 * 将列名列表转换为属性编号数组和包含它们的 Bitmapset；
 * 验证每个属性是否适合在发布列列表中（没有系统或生成的属性，没有重复）。
 * 后面会进行与副本身份的额外检查；见 check_publication_columns。
 *
 * 请注意，属性编号没有加上
 * FirstLowInvalidHeapAttributeNumber；系统列是禁止的，因此这是可以的。
 */
static void fc_publication_translate_columns(Relation fc_targetrel, List *fc_columns,
							  int *fc_natts, AttrNumber **fc_attrs)
{
	AttrNumber *fc_attarray = NULL;
	Bitmapset  *fc_set = NULL;
	ListCell   *fc_lc;
	int			fc_n = 0;
	TupleDesc	fc_tupdesc = RelationGetDescr(fc_targetrel);

	/* 当未定义列列表时退出。 */
	if (!fc_columns)
		return;

	/*
	 * 将列列表转换为 attnums。我们禁止系统属性，并确保没有重复的列。
	 */
	fc_attarray = palloc(sizeof(AttrNumber) * list_length(fc_columns));
	foreach(fc_lc, fc_columns)
	{
		char	   *fc_colname = strVal(lfirst(fc_lc));
		AttrNumber	fc_attnum = get_attnum(RelationGetRelid(fc_targetrel), fc_colname);

		if (fc_attnum == InvalidAttrNumber)
			ereport(ERROR,
					errcode(ERRCODE_UNDEFINED_COLUMN),
					errmsg("column \"%s\" of relation \"%s\" does not exist",
						   fc_colname, RelationGetRelationName(fc_targetrel)));

		if (!AttrNumberIsForUserDefinedAttr(fc_attnum))
			ereport(ERROR,
					errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
					errmsg("cannot use system column \"%s\" in publication column list",
						   fc_colname));

		if (TupleDescAttr(fc_tupdesc, fc_attnum - 1)->attgenerated)
			ereport(ERROR,
					errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
					errmsg("cannot use generated column \"%s\" in publication column list",
						   fc_colname));

		if (bms_is_member(fc_attnum, fc_set))
			ereport(ERROR,
					errcode(ERRCODE_DUPLICATE_OBJECT),
					errmsg("duplicate column \"%s\" in publication column list",
						   fc_colname));

		fc_set = bms_add_member(fc_set, fc_attnum);
		fc_attarray[fc_n++] = fc_attnum;
	}

	/* 保持整洁，以便目录表示始终排序 */
	qsort(fc_attarray, fc_n, sizeof(AttrNumber), fc_compare_int16);

	*fc_natts = fc_n;
	*fc_attrs = fc_attarray;

	bms_free(fc_set);
}

/*
 * 将列列表（由数组 Datum 表示）转换为 bitmapset。
 *
 * 如果 columns 不为 NULL，将列号添加到该集合中。
 *
 * 如果 mcxt 不为 NULL，在该上下文中构建 bitmapset。
 */
Bitmapset * pub_collist_to_bitmapset(Bitmapset *fc_columns, Datum fc_pubcols, MemoryContext fc_mcxt)
{
	Bitmapset  *fc_result = NULL;
	ArrayType  *fc_arr;
	int			fc_nelems;
	int16	   *fc_elems;
	MemoryContext fc_oldcxt = NULL;

	/*
	 * 如果提供了现有的 bitmap，则使用它。否则只使用 NULL 并构建一个新的 bitmap。
	 */
	if (fc_columns)
		fc_result = fc_columns;

	fc_arr = DatumGetArrayTypeP(fc_pubcols);
	fc_nelems = ARR_DIMS(fc_arr)[0];
	fc_elems = (int16 *) ARR_DATA_PTR(fc_arr);

	/* 如果指定了内存上下文，则切换到该上下文。 */
	if (fc_mcxt)
		fc_oldcxt = MemoryContextSwitchTo(fc_mcxt);

	for (int fc_i = 0; fc_i < fc_nelems; fc_i++)
		fc_result = bms_add_member(fc_result, fc_elems[fc_i]);

	if (fc_mcxt)
		MemoryContextSwitchTo(fc_oldcxt);

	return fc_result;
}

/*
 * 插入新的发布/架构映射。
 */
ObjectAddress publication_add_schema(Oid fc_pubid, Oid fc_schemaid, bool fc_if_not_exists)
{
	Relation	fc_rel;
	HeapTuple	fc_tup;
	Datum		fc_values[Natts_pg_publication_namespace];
	bool		fc_nulls[Natts_pg_publication_namespace];
	Oid			fc_psschid;
	Publication *fc_pub = GetPublication(fc_pubid);
	List	   *fc_schemaRels = NIL;
	ObjectAddress fc_myself,
				fc_referenced;

	fc_rel = table_open(PublicationNamespaceRelationId, RowExclusiveLock);

	/*
	 * 检查重复项。请注意，这实际上并不能真正防止重复项，它仅用于在常见情况下提供更好的错误信息。
	 * 真实的保护在于目录上的唯一键。
	 */
	if (SearchSysCacheExists2(PUBLICATIONNAMESPACEMAP,
							  ObjectIdGetDatum(fc_schemaid),
							  ObjectIdGetDatum(fc_pubid)))
	{
		table_close(fc_rel, RowExclusiveLock);

		if (fc_if_not_exists)
			return InvalidObjectAddress;

		ereport(ERROR,
				(errcode(ERRCODE_DUPLICATE_OBJECT),
				 errmsg("schema \"%s\" is already member of publication \"%s\"",
						get_namespace_name(fc_schemaid), fc_pub->name)));
	}

	fc_check_publication_add_schema(fc_schemaid);

	/* 形成一个元组 */
	memset(fc_values, 0, sizeof(fc_values));
	memset(fc_nulls, false, sizeof(fc_nulls));

	fc_psschid = GetNewOidWithIndex(fc_rel, PublicationNamespaceObjectIndexId,
								 Anum_pg_publication_namespace_oid);
	fc_values[Anum_pg_publication_namespace_oid - 1] = ObjectIdGetDatum(fc_psschid);
	fc_values[Anum_pg_publication_namespace_pnpubid - 1] =
		ObjectIdGetDatum(fc_pubid);
	fc_values[Anum_pg_publication_namespace_pnnspid - 1] =
		ObjectIdGetDatum(fc_schemaid);

	fc_tup = heap_form_tuple(RelationGetDescr(fc_rel), fc_values, fc_nulls);

	/* 将元组插入目录 */
	CatalogTupleInsert(fc_rel, fc_tup);
	heap_freetuple(fc_tup);

	ObjectAddressSet(fc_myself, PublicationNamespaceRelationId, fc_psschid);

	/* 添加对发布的依赖 */
	ObjectAddressSet(fc_referenced, PublicationRelationId, fc_pubid);
	recordDependencyOn(&fc_myself, &fc_referenced, DEPENDENCY_AUTO);

	/* 添加对模式的依赖 */
	ObjectAddressSet(fc_referenced, NamespaceRelationId, fc_schemaid);
	recordDependencyOn(&fc_myself, &fc_referenced, DEPENDENCY_AUTO);

	/* 关闭表 */
	table_close(fc_rel, RowExclusiveLock);

	/* 
	 * 使关系缓存无效，以便重建发布信息。详见
	 * publication_add_relation，了解我们为何需要考虑所有的
	 * 分区。
	 */
	fc_schemaRels = GetSchemaPublicationRelations(fc_schemaid,
											   PUBLICATION_PART_ALL);
	InvalidatePublicationRels(fc_schemaRels);

	return fc_myself;
}

/* 获取关系的发布 oid 列表 */
List * GetRelationPublications(Oid fc_relid)
{
	List	   *fc_result = NIL;
	CatCList   *fc_pubrellist;
	int			fc_i;

	/* 查找与该关系相关的所有发布。 */
	fc_pubrellist = SearchSysCacheList1(PUBLICATIONRELMAP,
									 ObjectIdGetDatum(fc_relid));
	for (fc_i = 0; fc_i < fc_pubrellist->n_members; fc_i++)
	{
		HeapTuple	fc_tup = &fc_pubrellist->members[fc_i]->tuple;
		Oid			fc_pubid = ((Form_pg_publication_rel) GETSTRUCT(fc_tup))->prpubid;

		fc_result = lappend_oid(fc_result, fc_pubid);
	}

	ReleaseSysCacheList(fc_pubrellist);

	return fc_result;
}

/* 
 * 获取发布的关系 oid 列表。
 *
 * 这只能用于 TABLE 发布，FOR ALL TABLES
 * 应使用 GetAllTablesPublicationRelations()。
 */
List * GetPublicationRelations(Oid fc_pubid, PublicationPartOpt fc_pub_partopt)
{
	List	   *fc_result;
	Relation	fc_pubrelsrel;
	ScanKeyData fc_scankey;
	SysScanDesc fc_scan;
	HeapTuple	fc_tup;

	/* 查找与该关系相关的所有发布。 */
	fc_pubrelsrel = table_open(PublicationRelRelationId, AccessShareLock);

	ScanKeyInit(&fc_scankey,
				Anum_pg_publication_rel_prpubid,
				BTEqualStrategyNumber, F_OIDEQ,
				ObjectIdGetDatum(fc_pubid));

	fc_scan = systable_beginscan(fc_pubrelsrel, PublicationRelPrpubidIndexId,
							  true, NULL, 1, &fc_scankey);

	fc_result = NIL;
	while (HeapTupleIsValid(fc_tup = systable_getnext(fc_scan)))
	{
		Form_pg_publication_rel fc_pubrel;

		fc_pubrel = (Form_pg_publication_rel) GETSTRUCT(fc_tup);
		fc_result = GetPubPartitionOptionRelations(fc_result, fc_pub_partopt,
												fc_pubrel->prrelid);
	}

	systable_endscan(fc_scan);
	table_close(fc_pubrelsrel, AccessShareLock);

	/* 现在对结果列表进行排序和去重 */
	list_sort(fc_result, list_oid_cmp);
	list_deduplicate_oid(fc_result);

	return fc_result;
}

/* 
 * 获取标记为 FOR ALL TABLES 的发布的 oid 列表。 */
List * GetAllTablesPublications(void)
{
	List	   *fc_result;
	Relation	fc_rel;
	ScanKeyData fc_scankey;
	SysScanDesc fc_scan;
	HeapTuple	fc_tup;

	/* 查找所有标记为所有表的发布。 */
	fc_rel = table_open(PublicationRelationId, AccessShareLock);

	ScanKeyInit(&fc_scankey,
				Anum_pg_publication_puballtables,
				BTEqualStrategyNumber, F_BOOLEQ,
				BoolGetDatum(true));

	fc_scan = systable_beginscan(fc_rel, InvalidOid, false,
							  NULL, 1, &fc_scankey);

	fc_result = NIL;
	while (HeapTupleIsValid(fc_tup = systable_getnext(fc_scan)))
	{
		Oid			fc_oid = ((Form_pg_publication) GETSTRUCT(fc_tup))->oid;

		fc_result = lappend_oid(fc_result, fc_oid);
	}

	systable_endscan(fc_scan);
	table_close(fc_rel, AccessShareLock);

	return fc_result;
}

/* 
 * 获取由 FOR ALL TABLES 发布的所有关系的列表。
 *
 * 如果该发布通过其各自的根分区表发布分区更改，
 * 我们必须排除分区，以便包含根分区表。
 */
#ifdef FDDM
List *fdd_get_exclude_schema(char *fc_fmmrExcludeSchemaStr)
{
	int   fc_len = 0;
	char  *fc_tmpStr = NULL;
	ListCell   *fc_lc = NULL;
	List *fc_result = NULL;
	List *fc_schama_list = NULL;
	if (strlen(fc_fmmrExcludeSchemaStr) == 0)
	{
		return NULL;
	}
	//解析schema，以逗号作为分隔
	fc_len = strlen(fc_fmmrExcludeSchemaStr) + 1;
	fc_tmpStr = palloc0(fc_len);	
	strncpy(fc_tmpStr, fc_fmmrExcludeSchemaStr, fc_len);

	if (!SplitIdentifierString(fc_tmpStr, ',', &fc_schama_list))
	{
		/* 列表中的语法错误 */
		pfree(fc_tmpStr);
		list_free(fc_schama_list);
		fc_schama_list = NULL;
		ereport(ERROR,
				(errcode(ERRCODE_SYNTAX_ERROR),
				 errmsg("invalid list syntax in parameter \"%s\"",
						"fmmr_exclude_schema")));
		return NULL;
	}

	foreach(fc_lc, fc_schama_list)
	{
		char* fc_schamename = (char *) lfirst(fc_lc);
		fc_result = lappend(fc_result, pstrdup(fc_schamename));
	}

	pfree(fc_tmpStr);
	return fc_result;
}

List * GetAllTablesPublicationRelations(bool fc_pubviaroot, bool fc_skipFddmSchema, bool fc_includePartition)
#else
List * GetAllTablesPublicationRelations(bool fc_pubviaroot)
#endif //FDDM
{
	Relation	fc_classRel;
	ScanKeyData fc_key[1];
	TableScanDesc fc_scan;
	HeapTuple	fc_tuple;
	List	   *fc_result = NIL;
#ifdef FDDM
	char 		*fc_conf_str = NULL;
	List 		*fc_excludeSchemaName = NIL;
	ListCell    *fc_lcname = NULL;
	List		*fc_excludeSchemaOid = NIL;

	if(fc_skipFddmSchema)
	{
		fc_conf_str = GetConfigOptionByName(FDD_EXCLUDE_SCHEMA_CONF, NULL, true);
		if (fc_conf_str != NULL)
		{
			fc_excludeSchemaName = fdd_get_exclude_schema(fc_conf_str);
		}
		foreach(fc_lcname, fc_excludeSchemaName)
		{
			char *fc_schemaname = (char*)lfirst(fc_lcname);
			Oid fc_schemaOid = get_namespace_oid(fc_schemaname, true);
			if(fc_schemaOid != InvalidOid)
			{
				fc_excludeSchemaOid = lappend_oid(fc_excludeSchemaOid, fc_schemaOid);
			}
		}
	}
#endif //FDDM

	fc_classRel = table_open(RelationRelationId, AccessShareLock);

	ScanKeyInit(&fc_key[0],
				Anum_pg_class_relkind,
				BTEqualStrategyNumber, F_CHAREQ,
				CharGetDatum(RELKIND_RELATION));

	fc_scan = table_beginscan_catalog(fc_classRel, 1, fc_key);

	while ((fc_tuple = heap_getnext(fc_scan, ForwardScanDirection)) != NULL)
	{
		Form_pg_class fc_relForm = (Form_pg_class) GETSTRUCT(fc_tuple);
		Oid			fc_relid = fc_relForm->oid;

		if (fc_is_publishable_class(fc_relid, fc_relForm) &&
#ifdef FDDM
			!list_member_oid(fc_excludeSchemaOid, fc_relForm->relnamespace) &&
#endif
			!(fc_relForm->relispartition && fc_pubviaroot))
			fc_result = lappend_oid(fc_result, fc_relid);
	}

	table_endscan(fc_scan);

#ifdef FDDM
	if (fc_pubviaroot || fc_includePartition)
#else
	if (fc_pubviaroot)
#endif
	{
		ScanKeyInit(&fc_key[0],
					Anum_pg_class_relkind,
					BTEqualStrategyNumber, F_CHAREQ,
					CharGetDatum(RELKIND_PARTITIONED_TABLE));

		fc_scan = table_beginscan_catalog(fc_classRel, 1, fc_key);

		while ((fc_tuple = heap_getnext(fc_scan, ForwardScanDirection)) != NULL)
		{
			Form_pg_class fc_relForm = (Form_pg_class) GETSTRUCT(fc_tuple);
			Oid			fc_relid = fc_relForm->oid;

			if (fc_is_publishable_class(fc_relid, fc_relForm) &&
#ifdef FDDM
				!list_member_oid(fc_excludeSchemaOid, fc_relForm->relnamespace) &&
#endif
				!fc_relForm->relispartition)
				fc_result = lappend_oid(fc_result, fc_relid);
		}

		table_endscan(fc_scan);
	}

	table_close(fc_classRel, AccessShareLock);
	return fc_result;
}

/* 
 * 获取发布的模式 oid 列表。
 *
 * 这只能用于 SCHEMA 中的 TABLES 发布。
 */
List * GetPublicationSchemas(Oid fc_pubid)
{
	List	   *fc_result = NIL;
	Relation	fc_pubschsrel;
	ScanKeyData fc_scankey;
	SysScanDesc fc_scan;
	HeapTuple	fc_tup;

	/* 查找与该发布相关的所有模式 */
	fc_pubschsrel = table_open(PublicationNamespaceRelationId, AccessShareLock);

	ScanKeyInit(&fc_scankey,
				Anum_pg_publication_namespace_pnpubid,
				BTEqualStrategyNumber, F_OIDEQ,
				ObjectIdGetDatum(fc_pubid));

	fc_scan = systable_beginscan(fc_pubschsrel,
							  PublicationNamespacePnnspidPnpubidIndexId,
							  true, NULL, 1, &fc_scankey);
	while (HeapTupleIsValid(fc_tup = systable_getnext(fc_scan)))
	{
		Form_pg_publication_namespace fc_pubsch;

		fc_pubsch = (Form_pg_publication_namespace) GETSTRUCT(fc_tup);

		fc_result = lappend_oid(fc_result, fc_pubsch->pnnspid);
	}

	systable_endscan(fc_scan);
	table_close(fc_pubschsrel, AccessShareLock);

	return fc_result;
}

/* 
 * 获取与指定模式相关的发布 oid 列表。 */
List * GetSchemaPublications(Oid fc_schemaid)
{
	List	   *fc_result = NIL;
	CatCList   *fc_pubschlist;
	int			fc_i;

	/* 查找与该模式相关的所有发布 */
	fc_pubschlist = SearchSysCacheList1(PUBLICATIONNAMESPACEMAP,
									 ObjectIdGetDatum(fc_schemaid));
	for (fc_i = 0; fc_i < fc_pubschlist->n_members; fc_i++)
	{
		HeapTuple	fc_tup = &fc_pubschlist->members[fc_i]->tuple;
		Oid			fc_pubid = ((Form_pg_publication_namespace) GETSTRUCT(fc_tup))->pnpubid;

		fc_result = lappend_oid(fc_result, fc_pubid);
	}

	ReleaseSysCacheList(fc_pubschlist);

	return fc_result;
}

/* 
 * 获取指定模式的可发布关系 oid 列表。 */
List * GetSchemaPublicationRelations(Oid fc_schemaid, PublicationPartOpt fc_pub_partopt)
{
	Relation	fc_classRel;
	ScanKeyData fc_key[1];
	TableScanDesc fc_scan;
	HeapTuple	fc_tuple;
	List	   *fc_result = NIL;

	Assert(OidIsValid(fc_schemaid));

	fc_classRel = table_open(RelationRelationId, AccessShareLock);

	ScanKeyInit(&fc_key[0],
				Anum_pg_class_relnamespace,
				BTEqualStrategyNumber, F_OIDEQ,
				fc_schemaid);

	/* 获取指定模式中存在的所有关系 */
	fc_scan = table_beginscan_catalog(fc_classRel, 1, fc_key);
	while ((fc_tuple = heap_getnext(fc_scan, ForwardScanDirection)) != NULL)
	{
		Form_pg_class fc_relForm = (Form_pg_class) GETSTRUCT(fc_tuple);
		Oid			fc_relid = fc_relForm->oid;
		char		fc_relkind;

		if (!fc_is_publishable_class(fc_relid, fc_relForm))
			continue;

		fc_relkind = get_rel_relkind(fc_relid);
		if (fc_relkind == RELKIND_RELATION)
			fc_result = lappend_oid(fc_result, fc_relid);
		else if (fc_relkind == RELKIND_PARTITIONED_TABLE)
		{
			List	   *fc_partitionrels = NIL;

			/* 
			 * 很可能其中一些分区处于与父表不同的
			 * 模式中，因此我们需要单独获取这些
			 * 分区。
			 */
			fc_partitionrels = GetPubPartitionOptionRelations(fc_partitionrels,
														   fc_pub_partopt,
														   fc_relForm->oid);
			fc_result = list_concat_unique_oid(fc_result, fc_partitionrels);
		}
	}

	table_endscan(fc_scan);
	table_close(fc_classRel, AccessShareLock);
	return fc_result;
}

/* 
 * 获取由 FOR TABLES IN SCHEMA
 * 发布的所有关系的列表。
 */
List * GetAllSchemaPublicationRelations(Oid fc_pubid, PublicationPartOpt fc_pub_partopt)
{
	List	   *fc_result = NIL;
	List	   *fc_pubschemalist = GetPublicationSchemas(fc_pubid);
	ListCell   *fc_cell;

	foreach(fc_cell, fc_pubschemalist)
	{
		Oid			fc_schemaid = lfirst_oid(fc_cell);
		List	   *fc_schemaRels = NIL;

		fc_schemaRels = GetSchemaPublicationRelations(fc_schemaid, fc_pub_partopt);
		fc_result = list_concat(fc_result, fc_schemaRels);
	}

	return fc_result;
}

/* 
 * 获取使用 oid 的发布
 *
 * Publication 结构及其数据在此进行 palloc。
 */
Publication *
GetPublication(Oid fc_pubid)
{
	HeapTuple	fc_tup;
	Publication *fc_pub;
	Form_pg_publication fc_pubform;

	fc_tup = SearchSysCache1(PUBLICATIONOID, ObjectIdGetDatum(fc_pubid));
	if (!HeapTupleIsValid(fc_tup))
		elog(ERROR, "cache lookup failed for publication %u", fc_pubid);

	fc_pubform = (Form_pg_publication) GETSTRUCT(fc_tup);

	fc_pub = (Publication *) palloc(sizeof(Publication));
	fc_pub->oid = fc_pubid;
	fc_pub->name = pstrdup(NameStr(fc_pubform->pubname));
	fc_pub->alltables = fc_pubform->puballtables;
	fc_pub->pubactions.pubinsert = fc_pubform->pubinsert;
	fc_pub->pubactions.pubupdate = fc_pubform->pubupdate;
	fc_pub->pubactions.pubdelete = fc_pubform->pubdelete;
	fc_pub->pubactions.pubtruncate = fc_pubform->pubtruncate;
	fc_pub->pubviaroot = fc_pubform->pubviaroot;

	ReleaseSysCache(fc_tup);

	return fc_pub;
}


/* 
 * 获取使用名称的发布。
 */
Publication *
GetPublicationByName(const char *fc_pubname, bool fc_missing_ok)
{
	Oid			fc_oid;

	fc_oid = get_publication_oid(fc_pubname, fc_missing_ok);

	return OidIsValid(fc_oid) ? GetPublication(fc_oid) : NULL;
}

/* 
 * get_publication_oid - 给定发布名称，查找 OID
 *
 * 如果 missing_ok 为假，找不到名称时抛出错误。如果为真，则
 * 仅返回 InvalidOid。
 */
Oid get_publication_oid(const char *fc_pubname, bool fc_missing_ok)
{
	Oid			fc_oid;

	fc_oid = GetSysCacheOid1(PUBLICATIONNAME, Anum_pg_publication_oid,
						  CStringGetDatum(fc_pubname));
	if (!OidIsValid(fc_oid) && !fc_missing_ok)
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("publication \"%s\" does not exist", fc_pubname)));
	return fc_oid;
}

/* 
 * get_publication_name - 给定发布 Oid，查找名称
 *
 * 如果 missing_ok 为假，找不到名称时抛出错误。如果为真，则
 * 仅返回 NULL。
 */
char * get_publication_name(Oid fc_pubid, bool fc_missing_ok)
{
	HeapTuple	fc_tup;
	char	   *fc_pubname;
	Form_pg_publication fc_pubform;

	fc_tup = SearchSysCache1(PUBLICATIONOID, ObjectIdGetDatum(fc_pubid));

	if (!HeapTupleIsValid(fc_tup))
	{
		if (!fc_missing_ok)
			elog(ERROR, "cache lookup failed for publication %u", fc_pubid);
		return NULL;
	}

	fc_pubform = (Form_pg_publication) GETSTRUCT(fc_tup);
	fc_pubname = pstrdup(NameStr(fc_pubform->pubname));

	ReleaseSysCache(fc_tup);

	return fc_pubname;
}

/* 返回发布中的表信息。 */
Datum pg_get_publication_tables(PG_FUNCTION_ARGS)
{
#define NUM_PUBLICATION_TABLES_ELEM	3
	FuncCallContext *fc_funcctx;
	char	   *fc_pubname = text_to_cstring(PG_GETARG_TEXT_PP(0));
	Publication *fc_publication;
	List	   *fc_tables;

	/* 仅在函数的第一次调用时完成的工作 */
	if (SRF_IS_FIRSTCALL())
	{
		TupleDesc	fc_tupdesc;
		MemoryContext fc_oldcontext;
#ifdef FDDM
		bool  fc_isFddPublication = false;

		if(strstr(fc_pubname, "fmmr_") == fc_pubname)
		{
			fc_isFddPublication = true;
		}
#endif //FDDM
		/* 为跨调用持久性创建一个函数上下文 */
		fc_funcctx = SRF_FIRSTCALL_INIT();

		/* 切换到适合多个函数调用的内存上下文 */
		fc_oldcontext = MemoryContextSwitchTo(fc_funcctx->multi_call_memory_ctx);

		fc_publication = GetPublicationByName(fc_pubname, false);

		/* 
		 * 发布支持分区表，尽管所有更改都
		 * 使用叶分区身份和模式进行复制，因此我们只
		 * 需要这些。
		 */
		if (fc_publication->alltables)
		{
#ifdef FDDM
			fc_tables = GetAllTablesPublicationRelations(fc_publication->pubviaroot, fc_isFddPublication, false);
#else
			fc_tables = GetAllTablesPublicationRelations(fc_publication->pubviaroot);
#endif
		}
		else
		{
			List	   *fc_relids,
					   *fc_schemarelids;

			fc_relids = GetPublicationRelations(fc_publication->oid,
											 fc_publication->pubviaroot ?
											 PUBLICATION_PART_ROOT :
											 PUBLICATION_PART_LEAF);
			fc_schemarelids = GetAllSchemaPublicationRelations(fc_publication->oid,
															fc_publication->pubviaroot ?
															PUBLICATION_PART_ROOT :
															PUBLICATION_PART_LEAF);
			fc_tables = list_concat_unique_oid(fc_relids, fc_schemarelids);

			/* 
			 * 如果发布通过其各自的根分区表发布分区更改，
			 * 我们必须排除分区以便包含根分区表。否则，
			 * 该函数可能会同时返回子表和父表，
			 * 这可能导致子表的数据在订阅者端被重复发布。
			 */
			if (fc_publication->pubviaroot)
				fc_tables = fc_filter_partitions(fc_tables);
		}

		/* 为结果行构造一个元组描述符。 */
		fc_tupdesc = CreateTemplateTupleDesc(NUM_PUBLICATION_TABLES_ELEM);
		TupleDescInitEntry(fc_tupdesc, (AttrNumber) 1, "relid",
						   OIDOID, -1, 0);
		TupleDescInitEntry(fc_tupdesc, (AttrNumber) 2, "attrs",
						   INT2VECTOROID, -1, 0);
		TupleDescInitEntry(fc_tupdesc, (AttrNumber) 3, "qual",
						   PG_NODE_TREEOID, -1, 0);

		fc_funcctx->tuple_desc = BlessTupleDesc(fc_tupdesc);
		fc_funcctx->user_fctx = (void *) fc_tables;

		MemoryContextSwitchTo(fc_oldcontext);
	}

	/* 每次调用函数时完成的工作 */
	fc_funcctx = SRF_PERCALL_SETUP();
	fc_tables = (List *) fc_funcctx->user_fctx;

	if (fc_funcctx->call_cntr < list_length(fc_tables))
	{
		HeapTuple	fc_pubtuple = NULL;
		HeapTuple	fc_rettuple;
		Oid			fc_relid = list_nth_oid(fc_tables, fc_funcctx->call_cntr);
		Oid			fc_schemaid = get_rel_namespace(fc_relid);
		Datum		fc_values[NUM_PUBLICATION_TABLES_ELEM];
		bool		fc_nulls[NUM_PUBLICATION_TABLES_ELEM];

		/*
		 * 使用适当的数据形成元组。
		 */
		MemSet(fc_nulls, 0, sizeof(fc_nulls));
		MemSet(fc_values, 0, sizeof(fc_values));

		fc_publication = GetPublicationByName(fc_pubname, false);

		fc_values[0] = ObjectIdGetDatum(fc_relid);

		/* 
		 * 我们不考虑 FOR ALL TABLES 或
		 * FOR TABLES IN SCHEMA 发布的行筛选器或列列表。
		 */
		if (!fc_publication->alltables &&
			!SearchSysCacheExists2(PUBLICATIONNAMESPACEMAP,
								   ObjectIdGetDatum(fc_schemaid),
								   ObjectIdGetDatum(fc_publication->oid)))
			fc_pubtuple = SearchSysCacheCopy2(PUBLICATIONRELMAP,
										   ObjectIdGetDatum(fc_relid),
										   ObjectIdGetDatum(fc_publication->oid));

		if (HeapTupleIsValid(fc_pubtuple))
		{
			/* 查找列列表属性。 */
			fc_values[1] = SysCacheGetAttr(PUBLICATIONRELMAP, fc_pubtuple,
										Anum_pg_publication_rel_prattrs,
										&(fc_nulls[1]));

			/* NULL 表示没有过滤器。 */
			fc_values[2] = SysCacheGetAttr(PUBLICATIONRELMAP, fc_pubtuple,
										Anum_pg_publication_rel_prqual,
										&(fc_nulls[2]));
		}
		else
		{
			fc_nulls[1] = true;
			fc_nulls[2] = true;
		}

		fc_rettuple = heap_form_tuple(fc_funcctx->tuple_desc, fc_values, fc_nulls);

		SRF_RETURN_NEXT(fc_funcctx, HeapTupleGetDatum(fc_rettuple));
	}

	SRF_RETURN_DONE(fc_funcctx);
}
